{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Hands-on!\n",
    "\n",
    "Nessa prática, sugerimos alguns pequenos exemplos para você implementar sobre o Spark."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Estimar o Pi\n",
    "\n",
    "Existe um algoritmo para estimar o Pi com números radômicos. Implemente-o sobre o Spark.\n",
    "\n",
    "Descrição do algoritmo: http://www.eveandersson.com/pi/monte-carlo-circle\n",
    "\n",
    "Implementação EM PYTHON (__não sobre o SPARK__): http://www.stealthcopter.com/blog/2009/09/python-calculating-pi-using-random-numbers/\n",
    "\n",
    "O númer de pontos deve ser 100000 (cem mill) vezes o número mínimo de partições padrão do seu SparkContext (`sc.defaultMinPartitions`). Esses pontos devem ser selecionados aleatóriamente na etapa de map (ver observações).\n",
    "\n",
    "Observações: use as funções __map__ (para mapear as ocorrêncas em `0` ou `1`, significando `1` quando o ponto aleatório cair dentro do círculo e `0` quando o contrário) e __reduce__ (para sumar as ocorrências)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3.164\n"
     ]
    }
   ],
   "source": [
    "from random import *\n",
    "from math import sqrt\n",
    "inside=0\n",
    "n=1000\n",
    "for i in range(0,n):\n",
    "    x=random()\n",
    "    y=random()\n",
    "    if sqrt(x*x+y*y)<=1:\n",
    "        inside+=1\n",
    "pi=4*inside/n\n",
    "\n",
    "print(pi)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "from random import *\n",
    "from math import sqrt\n",
    "def soma(a,b): return a+b\n",
    "def area(x,y):return 1 if sqrt(x*x+y*y)<=1 else  0\n",
    "def mapfunction(z):\n",
    "    x=random()\n",
    "    y=random()\n",
    "    return area(x,y)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "nummin = sc.defaultMinPartitions * 10000\n",
    "data = sc.parallelize(range(1, nummin))\n",
    "res1 = data.map(mapfunction)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "15767\n"
     ]
    }
   ],
   "source": [
    "print(res1.reduce(soma))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3.153557677883894\n"
     ]
    }
   ],
   "source": [
    "pispark = 4*(res1.reduce(soma))/(res1.count())\n",
    "print(pispark)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Filtragem de Primos\n",
    "\n",
    "Dado uma sequência de números de `1` a `1000000`, filtre somente os primos dessa sequência."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "def primes( n ):\n",
    "# i sera nosso divisor inicial\n",
    "    i = 1;\n",
    "# j sera nosso contador de ocorrências\n",
    "    j = 0;\n",
    "#Nenhum numero real vai ser divisivel por um numero maior do que sua metade\n",
    "    n1 = (n/2);\n",
    "\n",
    "    while (i <= n):\n",
    "\n",
    "       if (n % i==0):\n",
    "          i = i+1;\n",
    "          j = j+1;\n",
    "\n",
    "       if (i>=n1):\n",
    "        # damos a i, o valor da variavel entrada, pois o próximo divisor sera o próprio número\n",
    "          i = n;\n",
    "          i = i+1;\n",
    "          j = j+1;\n",
    "\n",
    "       else:\n",
    "          i = i+1;\n",
    "    if(j==2):\n",
    "        return True\n",
    "    else: \n",
    "        return False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[1, 2, 3, 4, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 67, 71, 73, 79, 83, 89, 97, 101, 103, 107, 109, 113, 127, 131, 137, 139, 149, 151, 157, 163, 167, 173, 179, 181, 191, 193, 197, 199, 211, 223, 227, 229, 233, 239, 241, 251, 257, 263, 269, 271, 277, 281, 283, 293, 307, 311, 313, 317, 331, 337, 347, 349, 353, 359, 367, 373, 379, 383, 389, 397, 401, 409, 419, 421, 431, 433, 439, 443, 449, 457, 461, 463, 467, 479, 487, 491, 499, 503, 509, 521]\n"
     ]
    }
   ],
   "source": [
    "data = sc.parallelize(range(1, 1000000))\n",
    "res = data.filter(primes)\n",
    "print(res.take(100))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Municípios do Brasil\n",
    "\n",
    "Dado o dataset `mucipios_do_Brasil.csv`, faça duas operações com ele:\n",
    "\n",
    "1. Monte uma lista dos municípios por estado.\n",
    "2. Conte quantos municípios há em cada estado.\n",
    "\n",
    "Dicas: use as operações groupByKey e reduceByKey, não faça um count na lista da operação 1. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "def mapfun(x):\n",
    "    return [x[0],x[1]]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[['uf', 'cidade'], ['AC', 'Acrelândia'], ['AC', 'Assis Brasil'], ['AC', 'Brasiléia'], ['AC', 'Bujari'], ['AC', 'Capixaba'], ['AC', 'Cruzeiro do Sul'], ['AC', 'Epitaciolândia'], ['AC', 'Feijó'], ['AC', 'Jordão']]\n"
     ]
    }
   ],
   "source": [
    "lines = sc.textFile(\"municipios_do_Brasil.csv\")\n",
    "\n",
    "data = lines.map(lambda line: line.split(\",\")).map(mapfun)\n",
    "\n",
    "print(data.take(10))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[['uf', 'cidade'], ['AC', 'Acrelândia'], ['AC', 'Assis Brasil'], ['AC', 'Brasiléia'], ['AC', 'Bujari'], ['AC', 'Capixaba'], ['AC', 'Cruzeiro do Sul'], ['AC', 'Epitaciolândia'], ['AC', 'Feijó'], ['AC', 'Jordão'], ['AC', 'Mâncio Lima'], ['AC', 'Manoel Urbano'], ['AC', 'Marechal Thaumaturgo'], ['AC', 'Plácido de Castro'], ['AC', 'Porto Acre'], ['AC', 'Porto Walter'], ['AC', 'Rio Branco'], ['AC', 'Rodrigues Alves'], ['AC', 'Santa Rosa do Purus'], ['AC', 'Sena Madureira']]\n"
     ]
    }
   ],
   "source": [
    "data.sortByKey()\n",
    "print(data.take(20))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('AC', ['Acrelândia', 'Assis Brasil', 'Brasiléia', 'Bujari', 'Capixaba', 'Cruzeiro do Sul', 'Epitaciolândia', 'Feijó', 'Jordão', 'Mâncio Lima', 'Manoel Urbano', 'Marechal Thaumaturgo', 'Plácido de Castro', 'Porto Acre', 'Porto Walter', 'Rio Branco', 'Rodrigues Alves', 'Santa Rosa do Purus', 'Sena Madureira', 'Senador Guiomard', 'Tarauacá', 'Xapuri'])]\n"
     ]
    }
   ],
   "source": [
    "Counts = data.groupByKey().map(lambda x : (x[0], list(x[1])))\n",
    "print(Counts.take(1))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('AC', 22), ('AP', 16), ('BA', 417), ('CE', 184)]\n"
     ]
    }
   ],
   "source": [
    "new = Counts.map(lambda tup: (tup[0], len(tup[1])))\n",
    "print(new.take(4))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## Word Count Memória Postumas de Brás Cubas\n",
    "\n",
    "Memórias Póstumas de Brás Cubas é um romance escrito por Machado de Assis, desenvolvido em princípio como folhetim, de março a dezembro de 1880, na Revista Brasileira, para, no ano seguinte, ser publicado como livro, pela então Tipografia Nacional.\n",
    "\n",
    "A obra retrata a escravidão, as classes sociais, o cientificismo e o positivismo da época. Dada essas informações, será que conseguimos idenficar essas características pelas palavras mais utilizadas em sua obra?\n",
    "\n",
    "Utilizando o dataset `Machado-de-Assis-Memorias-Postumas.txt`, faça um word count e encontre as palavras mais utilizadas por Machado de Assis em sua obra. Não esqueça de utilizar `stopwords.pt` para remover as `stop words`!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def filterfun(w):\n",
    "    stopwords = set(open('stopwords.pt').read().split())\n",
    "    if w[0] in stopwords:\n",
    "        return False\n",
    "    else:\n",
    "        return True"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "text_file = sc.textFile(\"Machado-de-Assis-Memorias-Postumas.txt\")\n",
    "\n",
    "counts = text_file.flatMap(lambda line: line.split(\" \")) \\\n",
    "             .map(lambda word: (word, 1)) \\\n",
    "             .reduceByKey(lambda a, b: a + b)\n",
    "counts = counts.filter(filterfun)\n",
    "counts = counts.takeOrdered(100, key = lambda x: -x[1])            "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('—', 723), ('', 303), ('Não', 210), ('O', 165), ('A', 163), ('lhe', 159), ('CAPÍTULO', 145), ('E', 117), ('Virgília', 109), ('olhos', 101), ('D.', 85), ('alguma', 81), ('Era', 76), ('Mas', 75), ('que,', 66), ('Quincas', 65), ('homem', 63), ('ia', 60), ('Que', 59), ('podia', 58), ('dizer', 57), ('Um', 55), ('Eu', 52), ('Lobo', 52), ('sei', 51), ('fui', 50), ('e,', 49), ('meus', 47), ('dizia', 47), ('logo', 47), ('algum', 45), ('mim', 45), ('idéia', 45), ('isto', 44), ('olhar', 44), ('eram', 44), ('lá', 43), ('talvez', 43), ('nossa', 41), ('Virgília,', 41), ('mão', 40), ('coisas', 40), ('fosse', 40), ('certo', 39), ('modo', 38), ('eu,', 38), ('No', 38), ('te', 37), ('tal', 37), ('tempo,', 36), ('Meu', 35), ('nada.', 35), ('depois,', 35), ('De', 35), ('ver', 35), ('pai', 34), ('Ao', 33), ('ir', 33), ('uns', 33), ('simples', 33), ('si', 32), ('algumas', 32), ('aí', 31), ('Talvez', 31), ('mim,', 31), ('Marcela', 31), ('É', 30), ('certa', 30), ('morte', 29), ('pé', 29), ('Se', 29), ('capítulo', 29), ('Brás', 28), ('anos,', 28), ('nosso', 28), ('Borba', 28), ('casa,', 28), ('muitas', 27), ('ele,', 27), ('filha', 27), ('Uma', 27), ('Já', 27), ('digo', 26), ('pai,', 26), ('boa', 26), ('disse-me', 26), ('porém,', 26), ('gesto', 26), ('Plácida', 26), ('verdade,', 25), ('Pois', 25), ('ali', 25), ('muito,', 25), ('porta', 25), ('homem,', 25), ('muita', 25), ('tu', 25), ('nunca', 25), ('coisa,', 25), ('nenhum', 24)]\n"
     ]
    }
   ],
   "source": [
    "print(counts)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "É dificil deduzir sobre o que o livro baseado apenas nas suas palavras. As palavras nos permitem deduzir que se trata de um livro(a palavra capítulo aparece no top 10), podemos também deduzir o nome de alguns personagens importantes como Quincas, Virgília, Brás e Marcela"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Apache Toree - PySpark",
   "language": "python",
   "name": "apache_toree_pyspark"
  },
  "language_info": {
   "codemirror_mode": "text/x-ipython",
   "file_extension": ".py",
   "mimetype": "text/x-ipython",
   "name": "python",
   "pygments_lexer": "python",
   "version": "3.6.3\n"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
